MetaGPT 课程思考题

1. MetaGPT 中 Agent 协作的核心机制

1. 消息传递系统

Agents 之间通过 Message 对象进行通信：

self.rc.env.publish_message(Message(content=card_content, cause_by=DealCards))

每个 Message 包含：

content: 消息内容
cause_by: 触发消息的 Action 类型
send_to: 指定接收者（可选，默认广播）

2. 事件监听机制

每个 Agent 通过 _watch 方法订阅特定类型的 Action 或消息：

# MathProdigy 角色监听 CallMathProdigy 动作
self._watch([CallMathProdigy])
 
# GameJudger 角色监听多种动作
self._watch([UserRequirement, RequireDealCardsAgain, HumanGiveExpression, MachineGiveExpression])

当环境中发布了匹配的消息时，监听该消息类型的 Agent 会被触发执行。

3. 动作执行流程

每个 Agent 通过 _act 方法定义其行为逻辑：

async def _act(self) -> Message:
    context = self.get_memories()
    msg = self.get_memories(k=1)[0]
    cause_by_str = f"{msg.cause_by}"
    
    # 根据触发消息类型执行不同逻辑
    if ("UserRequirement" in cause_by_str) or ("RequireDealCardsAgain" in cause_by_str):
        await self.call_deal_cards(context)
    else:
        # 执行其他逻辑...

4. 记忆与上下文共享

Agents 可以通过 get_memories() 访问历史消息，形成上下文感知：

context = self.get_memories()
point_list = get_recent_point_list(context)

24点游戏中的 Agent 协作流程

以 24 点游戏为例，三个 Agent 的协作流程如下：

初始化协作关系：

team = Team()
team.hire([MathProdigy(), GameJudger(), GamePlayer(is_human=True)])

触发协作链：
- GameJudger 发牌 → 发布 DealCards 消息
- GamePlayer 监听到 DealCards 消息 → 展示卡牌并获取玩家输入
- 根据玩家输入，GamePlayer 发布不同类型的消息：
  - HumanGiveExpression: 玩家提供表达式
  - CallMathProdigy: 玩家请求帮助
  - RequireDealCardsAgain: 玩家要求重新发牌
  - ExitGame: 玩家退出游戏
- GameJudger 监听到 HumanGiveExpression 消息 → 检查表达式正确性
- MathProdigy 监听到 CallMathProdigy 消息 → 提供正确表达式
反馈循环：
- 如果表达式错误，GameJudger 发布 WrongExpression 消息
- GamePlayer 监听到 WrongExpression 消息 → 提示玩家重试
- 如果表达式正确或需要重新发牌，循环继续

MetaGPT Agent 协作的关键特点

松耦合设计：
- Agents 不直接调用彼此的方法，而是通过消息通信
- 每个 Agent 只关注自己感兴趣的消息类型
事件驱动架构：
- 系统行为由消息事件驱动，而非固定的调用顺序
- 支持复杂的交互模式和条件分支
角色职责分离：
- 每个 Agent 有明确的职责和专长
- MathProdigy: 提供数学解决方案
- GameJudger: 管理游戏规则和判断
- GamePlayer: 处理用户交互
异步协作模式：
- 使用 async/await 支持非阻塞操作
- 允许 Agents 并行工作而不互相等待

这种基于消息的协作模式使 MetaGPT 应用能够构建复杂的多智能体系统，每个 Agent 专注于自己的专长领域，同时通过结构化的通信协议实现整体协作。

2. Action 与 Message 的关系及设计优点

在 MetaGPT 应用中，Action 与 Message 之间存在紧密的关系，且将 Action 作为可订阅事件的设计有多项优势。

Action 与 Message 的关系

1. 概念关系

Action: 表示 Agent 可以执行的具体行为或能力
Message: 表示 Agent 之间传递的信息载体

2. 结构关系

从代码中可以看到：

# Message 引用 Action 作为其来源
Message(content=card_content, cause_by=DealCards)
 
# Action 执行后产生内容，被封装到 Message 中
todo = DealCards()
card_content = await todo.run(context)
self.rc.env.publish_message(Message(content=card_content, cause_by=DealCards))

3. 功能关系

Action 是行为执行者：负责实际执行任务并产生结果

todo = CheckExpression()
check_result = await todo.run(msg.content)

Message 是信息承载者：负责传递 Action 的执行结果

self.rc.env.publish_message(Message(content=check_result, cause_by=WrongExpression))

Action 类型作为消息分类标识：Message 通过 cause_by 属性关联到特定 Action 类型

将 Action 作为可订阅事件的设计优点

1. 解耦与模块化

# 角色只需关注特定 Action，不需要知道谁发布了这些 Action
self._watch([CallMathProdigy])
self._watch([UserRequirement, RequireDealCardsAgain, HumanGiveExpression, MachineGiveExpression])

发布者与订阅者解耦：发布消息的 Agent 不需要知道谁会处理这个消息
功能模块化：每个 Action 代表一个独立功能单元，可以单独开发、测试和复用

2. 基于能力的协作模式

# 角色通过设置 Action 定义自己的能力
self.set_actions([MachineGiveExpression])
self.set_actions([DealCards])
self.set_actions([GetHumanReply])

能力导向：Agent 通过声明自己能处理哪些 Action 来定义自己的能力边界
专业分工：不同 Agent 可以专注于不同类型的 Action，形成专业分工

3. 声明式编程风格

# 通过声明监听关系，而非编写复杂的条件逻辑
class MathProdigy(Role):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._watch([CallMathProdigy])
        self.set_actions([MachineGiveExpression])

简化逻辑：不需要编写复杂的条件判断来决定谁处理什么消息
提高可读性：系统行为通过声明式的监听关系清晰表达

4. 事件驱动架构的灵活性

# 根据消息类型执行不同逻辑
cause_by_str = f"{msg.cause_by}"
if ("UserRequirement" in cause_by_str) or ("RequireDealCardsAgain" in cause_by_str):
    await self.call_deal_cards(context)
else:
    # 其他逻辑...

动态响应：系统可以根据运行时产生的事件动态调整行为
扩展性强：添加新功能只需创建新的 Action 类型和对应的处理逻辑

5. 行为追踪与调试便利

# 消息明确标记了其来源 Action
Message(content=human_reply, cause_by=HumanGiveExpression)

可追溯性：每个消息都带有其触发 Action 的信息，便于追踪系统行为
调试友好：可以通过监控特定类型的 Action 消息来调试系统

6. 支持复杂交互模式

# 一个 Action 可以触发多个后续 Action
if human_reply == "deal":
    self.rc.env.publish_message(Message(content="RequireDealCardsAgain", cause_by=RequireDealCardsAgain))
elif human_reply == "exit":
    self.rc.env.publish_message(Message(content="ExitGame", cause_by=ExitGame))
elif human_reply == "help":
    self.rc.env.publish_message(Message(content="CallMathProdigy", cause_by=CallMathProdigy))

多步骤工作流：支持构建复杂的多步骤、多分支工作流
条件触发：可以基于条件触发不同的 Action 链

7. 统一的交互接口

# 所有 Action 都遵循相同的接口模式
async def run(self, context: str):
    # 实现具体逻辑
    return result

一致性：所有 Action 遵循相同的接口约定，便于理解和使用
可组合性：不同 Action 可以方便地组合成更复杂的行为

这种设计将系统行为（Action）与通信机制（Message）分离但又紧密关联，既保持了概念清晰，又实现了灵活的事件驱动架构，特别适合构建复杂的多智能体系统。